package net.xuele.learn.rsocket;

import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.RSocketFactory;
import io.rsocket.transport.netty.client.TcpClientTransport;
import io.rsocket.util.DefaultPayload;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.commons.lang3.time.DateUtils;
import reactor.core.publisher.Flux;

import java.nio.ByteBuffer;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.List;

/**
 * @author: patrick-Hikari
 * @description: 不会收到服务端的response
 * @date: 2021/8/19 9:27 上午
 */
@Slf4j
public class FireNForgetClient {

    private final RSocket socket;

    private final List<Float> data;

    public FireNForgetClient() {
        this.socket = RSocketFactory.connect()
                .transport(TcpClientTransport.create("localhost",8070))
                .start()
                .block();
        // 准备好了数据
        this.data = Collections.unmodifiableList(generateData());
    }

    public void sendData() {
        // 每隔50ms发送一次数据
        Flux.interval(Duration.ofMillis(50))
                .take(data.size())
                .map(this::createFloatPayload)
                .flatMap(socket::fireAndForget)
                .blockLast();
    }

    private Payload createFloatPayload(Long index) {
        System.out.println("before..." + DateFormatUtils.ISO_DATETIME_TIME_ZONE_FORMAT.format(new Date()));
        float velocity = data.get(index.intValue());
        ByteBuffer buffer = ByteBuffer.allocate(4).putFloat(velocity);
        buffer.rewind();
        return DefaultPayload.create(buffer);
    }

    private List<Float> generateData() {
        List<Float> dataList = new ArrayList<>(30);
        float velocity = 0;
        for (int i = 0; i < 30; i++) {
            velocity += Math.random();
            dataList.add(velocity);
        }
        return dataList;
    }

    public List<Float> getData() {
        return data;
    }

    public void dispose() {
        this.socket.dispose();
    }
}
